package com.gzyj.flink.gps.map.beidoualarm;

import cn.hutool.core.collection.CollectionUtil;
import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.alibaba.fastjson.serializer.SerializerFeature;
import com.alibaba.fastjson.support.spring.FastJsonRedisSerializer;
import com.gzyj.flink.driver.Driverlogininfo;
import com.gzyj.flink.dto.alarm.AlarmALLObject;
import com.gzyj.flink.dto.alarm.Kafka_VehicleNormalAlarm;
import com.gzyj.flink.dto.alarm.MonitorClientAlarm;
import com.gzyj.flink.dto.utils.ToolUtils;
import com.gzyj.flink.gps.config.RedisKeyPrefixConfig;
import com.gzyj.flink.gps.factory.MySQLMybatisSessionFactory;
import com.gzyj.flink.gps.factory.RedisSessionFactory;
import com.gzyj.flink.gps.mapper.mysql.MonYanzhongweiguiSetMapper;
import com.gzyj.flink.gps.utils.TimeUtils;
import com.gzyj.flink.pojo.gps.BaobiaoAlarmsummary;
import com.gzyj.flink.pojo.mysql.set.MonYanzhongweiguiSet;
import com.gzyj.flink.veh.VehicleIncrement;
import org.apache.commons.lang3.StringUtils;
import org.apache.flink.api.common.functions.RichFlatMapFunction;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.util.Collector;
import org.apache.ibatis.session.SqlSession;
import org.apache.ibatis.session.SqlSessionFactory;
import org.springframework.data.redis.connection.lettuce.LettuceConnectionFactory;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.data.redis.serializer.StringRedisSerializer;

import java.math.BigDecimal;
import java.util.*;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.TimeUnit;

/**
 * 北斗终端报警数据处理
 */
public class BeidouZongduanDataProcess extends RichFlatMapFunction<Kafka_VehicleNormalAlarm, Object> {

    SqlSessionFactory factory;
    LettuceConnectionFactory redisFactory;

    @Override
    public void open(Configuration parameters) throws Exception {
        factory = MySQLMybatisSessionFactory.getSqlSessionFactory();
        redisFactory = RedisSessionFactory.getRedisFactory();
    }

    private RedisTemplate<String, Object> getRedisTemplate(LettuceConnectionFactory redisSessionFactory) {
        RedisTemplate<String, Object> redisTemplate = new RedisTemplate<>();
        redisTemplate.setConnectionFactory(redisSessionFactory);
        redisTemplate.setKeySerializer(new StringRedisSerializer());
        redisTemplate.setValueSerializer(new FastJsonRedisSerializer<>(Object.class));
        redisTemplate.setHashKeySerializer(new StringRedisSerializer());
        redisTemplate.setHashValueSerializer(new FastJsonRedisSerializer<>(Object.class));
        redisTemplate.afterPropertiesSet();
        return redisTemplate;
    }

    public VehicleIncrement getVehicleInfo(String vehId, RedisTemplate<String, Object> clusterRedisTemplate) {

        Set<String> keys = clusterRedisTemplate.keys(RedisKeyPrefixConfig.RedisKey_VehicleInfo + "*:" + vehId + "*");
        VehicleIncrement vehicleInfo = null;
        if (CollectionUtil.isNotEmpty(keys)) {
            Object obj = clusterRedisTemplate.opsForValue().get(new ArrayList<String>(keys).get(0));
            if (obj == null) {
                return null;
            } else {
                vehicleInfo = JSONObject.parseObject(JSON.toJSONString(obj, SerializerFeature.DisableCircularReferenceDetect), VehicleIncrement.class);
            }
        } else {
            return null;
        }
        return vehicleInfo;
    }
    public Driverlogininfo TryGetData_DLD(String vehId, RedisTemplate<String, Object> clusterRedisTemplate) {
        Object obj = clusterRedisTemplate.opsForValue().get(RedisKeyPrefixConfig.RedisKey_DriverSignOn + vehId);
        if (obj == null) {
            return new Driverlogininfo();
        } else {
            return JSON.parseObject(JSON.toJSONString(obj, SerializerFeature.DisableCircularReferenceDetect), Driverlogininfo.class);
        }

    }

    public String trimend(String inStr, String suffix) {
        while (inStr.endsWith(suffix)) {
            inStr = inStr.substring(0, inStr.length() - suffix.length());
        }
        return inStr;
    }

    /**
     * @param VehAlarm  当前报警参数
     * @param collector
     * @throws Exception
     */
    @Override
    public void flatMap(Kafka_VehicleNormalAlarm VehAlarm, Collector<Object> collector) throws Exception {

        SqlSession session = factory.openSession(true);
        List<Object> list = new ArrayList<>();
        try {
            MonYanzhongweiguiSetMapper monYanzhongweiguiSetMapper = session.getMapper(MonYanzhongweiguiSetMapper.class);

            MonYanzhongweiguiSet AlarmDeploy = monYanzhongweiguiSetMapper.selectByVehId(VehAlarm.getVehicleId());
            session.commit();
            RedisTemplate<String, Object> clusterRedisTemplate = getRedisTemplate(redisFactory);
            VehicleIncrement vehinfo = getVehicleInfo(VehAlarm.getVehicleId(), clusterRedisTemplate);

            Map<Object, Object> tmp = clusterRedisTemplate.opsForHash().entries(RedisKeyPrefixConfig.RedisKey_CmAlarm + VehAlarm.getVehicleId());
            // 获取开始报警数据
            ConcurrentHashMap<Object, Object> VehicleAlarm = new ConcurrentHashMap<>(tmp);

            String[] sNote = trimend(VehAlarm.getNote(), ",").split(",");
            Driverlogininfo driverlogininfo=TryGetData_DLD(VehAlarm.getVehicleId(), clusterRedisTemplate);

            if (StringUtils.isNotEmpty(VehAlarm.getNote())) {
                for (int i = 0; i < sNote.length; i++) {
                    String sAlarmNote = sNote[i];
                    if (sAlarmNote.equalsIgnoreCase("疲劳驾驶")) sAlarmNote = "疲劳驾驶报警";
                    sNote[i] = sAlarmNote;
                    if (VehicleAlarm.containsKey(sAlarmNote)) {
                        AlarmALLObject Lastalarm = JSON.parseObject((String) VehicleAlarm.get(sAlarmNote), AlarmALLObject.class);

                        // 结束报警
                        double KeepTime = TimeUtils.getMinusSecond(VehAlarm.getTime(), Lastalarm.BeginTime);
                        int ChaoSuBi = 0;
                        int status = 1;

                        if (KeepTime < 0) KeepTime = 0;

                        // 疲劳驾驶严重违规计算
                        if (sAlarmNote.equalsIgnoreCase("疲劳驾驶报警")) {
                            int plsj = -1;
                            int plsj1 = -1;
                            int plsj2 = -1;
                            if (AlarmDeploy != null) {


                                plsj = AlarmDeploy.getPlcxsj() == null ? -1 : AlarmDeploy.getPlcxsj();
                                plsj1 = AlarmDeploy.getPlcxsj1() == null ? -1 : AlarmDeploy.getPlcxsj1();
                                plsj2 = AlarmDeploy.getPlcxsj2() == null ? -1 : AlarmDeploy.getPlcxsj2();

                                if (plsj1 >= 0) {
                                    plsj1 = plsj1 * 60;
                                    if (KeepTime >= plsj1) status = 1;
                                }
                                if (plsj2 >= 0) {
                                    plsj2 = plsj2 * 60;
                                    if (KeepTime >= plsj2) status = 2;
                                }
                                if (plsj >= 0) {
                                    plsj = plsj * 60;
                                    if (KeepTime >= plsj) status = 3;
                                }
                            }
                        }


                        // 超速报警严重违规计算
                        if (sAlarmNote.equalsIgnoreCase("超速报警")) {
                            int cssj = -1;
                            int cssj1 = -1;
                            int cssj2 = -1;
                            double csbfb = -1;
                            double csbfb1 = -1;
                            double csbfb2 = -1;
                            if (AlarmDeploy != null) {
                                //  计算严重违规等级 报警等级: status 默认为1级

                                cssj = AlarmDeploy.getCscxsj() == null ? -1 : AlarmDeploy.getCscxsj();
                                cssj1 = AlarmDeploy.getCscxsj1() == null ? -1 : AlarmDeploy.getCscxsj1();
                                cssj2 = AlarmDeploy.getCscxsj2() == null ? -1 : AlarmDeploy.getCscxsj2();
                                csbfb = AlarmDeploy.getCsbfb() == null ? -1 : AlarmDeploy.getCsbfb();
                                csbfb1 = AlarmDeploy.getCsbfb1() == null ? -1 : AlarmDeploy.getCsbfb1();
                                csbfb2 = AlarmDeploy.getCsbfb2() == null ? -1 : AlarmDeploy.getCsbfb2();

                                if (cssj1 >= 0 && csbfb1 >= 0) {
                                    if (KeepTime >= cssj1 || ChaoSuBi >= csbfb1) status = 2;
                                }
                                if (cssj2 >= 0 && csbfb2 >= 0) {
                                    if (KeepTime >= cssj2 || ChaoSuBi >= csbfb2) status = 3;
                                }

                            }
                        }

                        BaobiaoAlarmsummary baobiaoAlarmsummary = new BaobiaoAlarmsummary();
                        baobiaoAlarmsummary.setVehid(VehAlarm.getVehicleId());
                        baobiaoAlarmsummary.setDeptId(VehAlarm.getDeptId());
                        baobiaoAlarmsummary.setBegintime(Lastalarm.BeginTime);
                        baobiaoAlarmsummary.setDriverName(driverlogininfo.getDriverName());
                        baobiaoAlarmsummary.setDriverNo(driverlogininfo.getDriverNo());
                        baobiaoAlarmsummary.setAlarmtype(sAlarmNote);
                        baobiaoAlarmsummary.setAnalyzemode(0);

                        baobiaoAlarmsummary.setMaxspeed(Lastalarm.MaxSpeed);

                        baobiaoAlarmsummary.setLongitude(BigDecimal.valueOf(Lastalarm.BeginLon));
                        baobiaoAlarmsummary.setLatitude(BigDecimal.valueOf(Lastalarm.BeginLat));
                        baobiaoAlarmsummary.setVelocity(VehAlarm.getVelocity());
                        baobiaoAlarmsummary.setAngle(VehAlarm.getAngle());
                        baobiaoAlarmsummary.setLocal(VehAlarm.getLocal() ? 1 : 0);
                        baobiaoAlarmsummary.setElevation(BigDecimal.ZERO);
                        baobiaoAlarmsummary.setLimited(VehAlarm.getLimited());

                        if (Lastalarm.getBeginSpeed() > 30 && KeepTime < 50000) {

                            baobiaoAlarmsummary.setPassed(100);

                        } else {
                            if (sAlarmNote.equalsIgnoreCase("超速报警") || sAlarmNote.equalsIgnoreCase("疲劳驾驶报警")) {

                                baobiaoAlarmsummary.setPassed(-1);
                            } else {
                                baobiaoAlarmsummary.setPassed(100);
                            }
                        }

                        baobiaoAlarmsummary.setTime(Lastalarm.BeginTime);
                        baobiaoAlarmsummary.setKeeptime(KeepTime);
                        baobiaoAlarmsummary.setRoadName(Lastalarm.getBeginRoadName());
                        baobiaoAlarmsummary.setRoadLevel("");
                        baobiaoAlarmsummary.setRoadLimited(0);
                        baobiaoAlarmsummary.setDisposealarmname("");
                        baobiaoAlarmsummary.setDisposealarmtime(Lastalarm.BeginTime);

                        baobiaoAlarmsummary.setVerifyname("System");
                        baobiaoAlarmsummary.setVerifytime(TimeUtils.toDate(TimeUtils.getNow()));

                        baobiaoAlarmsummary.setStatus(status);
                        baobiaoAlarmsummary.setSyntime(TimeUtils.toDate(TimeUtils.getNow()));
                        baobiaoAlarmsummary.setPlatenumber(vehinfo.getCheliangpaizhao());
                        baobiaoAlarmsummary.setColor(vehinfo.getChepaiyanse());
                        baobiaoAlarmsummary.setCompany(vehinfo.getDeptName());
                        baobiaoAlarmsummary.setOperattype(vehinfo.getShiyongxingzhi());
                        baobiaoAlarmsummary.setDistance(Lastalarm.getDistance());

                        baobiaoAlarmsummary.setIsregionv("0");
                        baobiaoAlarmsummary.setAtnocturnal(0);
                        baobiaoAlarmsummary.setBaojingtype(1);
                        baobiaoAlarmsummary.setChaosubi(ChaoSuBi);
                        baobiaoAlarmsummary.setAlarmguid(Lastalarm.getAlarmGuid());

                        String sDriver = VehAlarm.getDriver() == null ? "" : VehAlarm.getDriver();
                        baobiaoAlarmsummary.setIcardsign(sDriver.length() <= 50 ? sDriver : sDriver.substring(0, 50));

                        list.add(baobiaoAlarmsummary);


                        if (sAlarmNote.equalsIgnoreCase("超速报警") || sAlarmNote.equalsIgnoreCase("疲劳驾驶报警")) {
                            //推送到kafka
                            MonitorClientAlarm vi = new MonitorClientAlarm();
                            vi.VehNo = vehinfo.getCheliangpaizhao();
                            vi.deptName = vehinfo.getDeptName();
                            vi.VehId = VehAlarm.getVehicleId();
                            vi.PlateColor = vehinfo.getChepaiyanse();
                            vi.AlarmFlag = 1;
                            vi.Time = VehAlarm.getTime();
                            vi.Note = sAlarmNote;
                            vi.Longitude = VehAlarm.getLongitude();
                            vi.Latitude = VehAlarm.getLatitude();
                            vi.Velocity = VehAlarm.getVelocity();
                            vi.Angle = VehAlarm.getAngle();
                            vi.Local = VehAlarm.getLocal() ? 1 : 0;
                            vi.Limited = 0;
                            vi.RoadLimited = 0;
                            vi.RoadName = VehAlarm.getRoadName();
                            vi.AlarmGuid = Lastalarm.getAlarmGuid();
                            vi.Driver = VehAlarm.getDriver();
                            vi.alarmSource = 2;
                            list.add(vi);
                        }


                    } else {

                        System.out.println("不包含 需要添加");

                    }
                }

                VehAlarm.setNote(String.join(",", sNote));
                Iterator<Map.Entry<Object, Object>> mapiter = VehicleAlarm.entrySet().iterator();
                while (mapiter.hasNext()) {
                    Map.Entry<Object, Object> entry = mapiter.next();
                    String sAlarmNote = (String) entry.getKey();
                    if (sAlarmNote.equalsIgnoreCase("疲劳驾驶")) sAlarmNote = "疲劳驾驶报警";
                    if (!VehAlarm.getNote().contains(sAlarmNote)) {
                        // 结束报警

                        AlarmALLObject Lastalarm = JSON.parseObject((String) entry.getValue(), AlarmALLObject.class);
                        BaobiaoAlarmsummary baobiaoAlarmsummary=new BaobiaoAlarmsummary();
                        baobiaoAlarmsummary.setVehid(VehAlarm.getVehicleId());
                        baobiaoAlarmsummary.setDeptId(VehAlarm.getDeptId());
                        baobiaoAlarmsummary.setBegintime(Lastalarm.getBeginTime());
                        baobiaoAlarmsummary.setAlarmtype(sAlarmNote);
                        baobiaoAlarmsummary.setAnalyzemode(0);

                        baobiaoAlarmsummary.setEndtime(VehAlarm.getTime());
                        baobiaoAlarmsummary.setEndspeed(VehAlarm.getVelocity());
                        baobiaoAlarmsummary.setCutofftime(TimeUtils.toDate(TimeUtils.getNow()));
                        baobiaoAlarmsummary.setEndlongitude(BigDecimal.valueOf(VehAlarm.getLongitude()));
                        baobiaoAlarmsummary.setEndlatitude(BigDecimal.valueOf(VehAlarm.getLatitude()));
                        baobiaoAlarmsummary.setEndroadname("未知道路");
                        double KeepTime = TimeUtils.getMinusSecond(VehAlarm.getTime(), Lastalarm.BeginTime);
                        baobiaoAlarmsummary.setKeeptime(KeepTime);
                        baobiaoAlarmsummary.setDistance(ToolUtils.GetDistance(VehAlarm.getLongitude(),VehAlarm.getLatitude(),Lastalarm.getBeginLon(),Lastalarm.getBeginLat()));

                        list.add(baobiaoAlarmsummary);

                        clusterRedisTemplate.opsForHash().delete(RedisKeyPrefixConfig.RedisKey_CmAlarm + VehAlarm.getVehicleId(), sAlarmNote);

                    }
                }


            } else {
                for (int i = 0; i < sNote.length; i++) {
                    String sAlarmNote = sNote[i];
                    if (sAlarmNote.equalsIgnoreCase("疲劳驾驶")) sAlarmNote = "疲劳驾驶报警";
                    sNote[i] = sAlarmNote;
                }

                VehAlarm.setNote(String.join(",", sNote));
                Iterator<Map.Entry<Object, Object>> mapiter = VehicleAlarm.entrySet().iterator();

                while (mapiter.hasNext()) {
                    Map.Entry<Object, Object> entry = mapiter.next();

                    String sAlarmNote = (String) entry.getKey();
                    if (sAlarmNote.equalsIgnoreCase("疲劳驾驶")) sAlarmNote = "疲劳驾驶报警";

                    AlarmALLObject Lastalarm = JSON.parseObject((String) entry.getValue(), AlarmALLObject.class);
                    BaobiaoAlarmsummary baobiaoAlarmsummary=new BaobiaoAlarmsummary();
                    baobiaoAlarmsummary.setVehid(VehAlarm.getVehicleId());
                    baobiaoAlarmsummary.setDeptId(VehAlarm.getDeptId());
                    baobiaoAlarmsummary.setBegintime(Lastalarm.getBeginTime());
                    baobiaoAlarmsummary.setAlarmtype(sAlarmNote);
                    baobiaoAlarmsummary.setAnalyzemode(0);

                    baobiaoAlarmsummary.setEndtime(VehAlarm.getTime());
                    baobiaoAlarmsummary.setEndspeed(VehAlarm.getVelocity());
                    baobiaoAlarmsummary.setCutofftime(TimeUtils.toDate(TimeUtils.getNow()));
                    baobiaoAlarmsummary.setEndlongitude(BigDecimal.valueOf(VehAlarm.getLongitude()));
                    baobiaoAlarmsummary.setEndlatitude(BigDecimal.valueOf(VehAlarm.getLatitude()));
                    baobiaoAlarmsummary.setEndroadname("未知道路");
                    double KeepTime = TimeUtils.getMinusSecond(VehAlarm.getTime(), Lastalarm.BeginTime);
                    baobiaoAlarmsummary.setKeeptime(KeepTime);
                    baobiaoAlarmsummary.setDistance(ToolUtils.GetDistance(VehAlarm.getLongitude(),VehAlarm.getLatitude(),Lastalarm.getBeginLon(),Lastalarm.getBeginLat()));

                    list.add(baobiaoAlarmsummary);


                }


                //结束所有报警
                clusterRedisTemplate.delete(RedisKeyPrefixConfig.RedisKey_CmAlarm + VehAlarm.getVehicleId());


            }


        } catch (Exception e) {
            session.rollback();

        } finally {

            session.close();

            for (Object item : list) {
                collector.collect(item);
            }


        }


    }


}
